package com.bj58.ecdata.hadoop.fourthinfo;

import java.io.BufferedReader;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.Serializable;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.Arrays;
import java.util.Date;
import java.util.HashMap;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.filecache.DistributedCache;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;


public class FourthInfo {
	private static final String input="/dsap/resultdata/infostat/ZP_InfoJobHDFS/yyyyMMdd/";
	private static final String output="/dsap/resultdata/ec_data/EC_Post_Detail_Job/yyyyMMdd/";

	// Mapper
	public static class FourthInfoMapper extends Mapper<LongWritable, Text, Text, Text> {
		private static Pattern p = Pattern.compile("^cate(first|second)(\\d+)_local(first|second|third)(\\d+)_(.*?)(\\d{8})_([iu]count:\\d+)");
		private Text mapOutputKey = new Text();
		private Text mapOutputValue = new Text();
		public void map(LongWritable key, Text value, Context context) {
			Matcher m = p.matcher(value.toString());
			if (m.find()) {
				String cateid_localid = m.group(2) + "_" + m.group(4);
				String info = m.group(5) + m.group(7);
				mapOutputKey.set(cateid_localid);
				mapOutputValue.set(info);
				try {
					context.write(mapOutputKey, mapOutputValue);
				} catch (Exception e) {
					e.printStackTrace();
					return;
				}
			}
		}
	}

	// Reducer
	public static class FourthInfoReducer extends Reducer<Text, Text, NullWritable, Text> {
		private String[] fields = { "cateid","stat_date", "local_id","dispcity1", "dispcity2",
			"dispcity3", "cate_id","belong_cate1", "belong_cate2", "total_info_count",
			"total_info_users", "valid_info_count", "valid_info_users",
			"new_info_count", "new_info_users", "modify_info_count",
			"modify_info_users", "refresh_info_count", "refresh_info_users",
			"new_valid_info_count", "new_valid_info_users",
			"ordinary_valid_info_count", "ordinary_valid_info_users",
			"ordinary_new_info_count", "ordinary_new_info_users",
			"ordinary_modify_info_count", "ordinary_modify_info_users",
			"ordinary_refresh_info_count", "ordinary_refresh_info_users",
			"ordinary_new_valid_info_count", "ordinary_new_valid_info_users",
			"vip_valid_info_count", "vip_valid_info_users",
			"vip_new_info_count", "vip_new_info_users",
			"vip_modify_info_count", "vip_modify_info_users",
			"vip_refresh_info_count", "vip_refresh_info_users",
			"vip_new_valid_info_count", "vip_new_valid_info_users" };
		private HashMap<String, Integer> fieldsIndex = null;
		private HashMap<String, String> dict = new HashMap<String, String>();
		private HashMap<String,LocalInfo> localInfos=null;
		private HashMap<String,CateInfo> cateInfos=null;
		private String date;
		private NullWritable nullKey=NullWritable.get();
		private Text outputValue=new Text();
		public void setup(Context context) {
			String[] states = { "state1", "uptype0", "uptype1", "uptype2",
			"state1_uptype0" };
			String[] types = { "valid", "new", "modify", "refresh", "new_valid" };

			String[] src = { "", "source0", "source6" };
			String[] roles = { "", "ordinary", "vip" };

			String[] countType = { "_info_count", "_info_users" };
			String[] counters = { "icount", "ucount" };
			for (int i = 0; i < types.length; i++)
				for (int j = 0; j < roles.length; j++)
					for (int k = 0; k < countType.length; k++) {
						if(j==0)
							dict.put(states[i] + "_" + counters[k], types[i] +  countType[k]);
						else
							dict.put(states[i] + "_" + src[j] + "_" + counters[k],roles[j] + "_" + types[i] +  countType[k]);
					}
			dict.put("icount", "total_info_count");
			dict.put("ucount", "total_info_users");
			fieldsIndex = new HashMap<String, Integer>();
			for (int i = 0; i < fields.length; i++) {
				fieldsIndex.put(fields[i], i);
			}
			date = context.getConfiguration().get("DATE");
			try {
				BufferedReader br = new BufferedReader(new InputStreamReader(new FileInputStream("BelongCate.part"),"UTF-8"));
				cateInfos=getCateInfo(br);
				br = new BufferedReader(new InputStreamReader(new FileInputStream("DisLocal.part"),"UTF-8"));
				localInfos=getLocalInfo(br);
				System.out.println(localInfos.size()+" localInfo!");
				System.out.println(cateInfos.size()+" cateInfo!");
			} catch (IOException e) {
				e.printStackTrace();
				return;
			}
		}

		public void reduce(Text key, Iterable<Text> values, Context context) {
			String[] cate_local_id=key.toString().split("_");
			String cateid=cate_local_id[0]; 
			String localid=cate_local_id[1];
			if(!cateInfos.containsKey(cateid)||!localInfos.containsKey(localid))
					return;
			String[] row=new String[fields.length];
			boolean flag=false;
			Arrays.fill(row, "0");
			row[0]=cateid;
			row[1]=date;
			row[fieldsIndex.get("cate_id")]=cateid;
			row[fieldsIndex.get("local_id")]=localid;
			CateInfo  cate=cateInfos.get(cateid);
			String[] catesIds=cate.getCateids();
			int index=fieldsIndex.get("belong_cate1");
			for(int i=0;i<2;i++){
				if(catesIds[i].equals("0")||!cateInfos.containsKey(catesIds[i]))
					row[index+i]="";
				else
					row[index+i]=cateInfos.get(catesIds[i]).getCatename();
			}
			LocalInfo local=localInfos.get(localid);
			String[] localIds=local.getLocalids();
			index=fieldsIndex.get("dispcity1");
			for(int i=0;i<3;i++){
				if(localIds[i].equals("0")||!localInfos.containsKey(localIds[i]))
					row[index+i]="";
				else
					row[index+i]=localInfos.get(localIds[i]).getLocalname();
			}	
			for(Text t:values){
				String[] info=t.toString().split(":");
				String infoKey=info[0].trim();
				String infoValue=info[1].trim();
				String f=dict.get(infoKey);
				if(f!=null){
					flag=true;
					int fieldindex=fieldsIndex.get(f);
					 row[fieldindex]=infoValue;
				}
			}
			if (flag == false)
				return;
			StringBuilder strbuilder = new StringBuilder();
			for (int i = 0; i < fields.length - 1; i++) {
				strbuilder.append(row[i] + ",");
			}
			strbuilder.append(row[row.length - 1]);
			outputValue.set(strbuilder.toString());
			try {
				context.write(nullKey, outputValue);
			} catch (Exception e) {
				e.printStackTrace();
				return;
			}
		}
	
		public HashMap<String,LocalInfo> getLocalInfo(BufferedReader reader) throws IOException{
			String line=null;
			HashMap<String,LocalInfo> locals=new HashMap<String,LocalInfo>();
			while((line=reader.readLine())!=null){
				String fields[]=line.split("\t");
				String localid=fields[0];
				String localname=fields[2];
				LocalInfo local=new LocalInfo(localname);
				String levels[]=fields[4].split(",");
				int len=levels.length<3?levels.length:3;
				for(int i=0;i<len;i++){
					local.setLocalids(i, levels[i]);
				}
				locals.put(localid, local);
			}
			reader.close();
			return locals;
		}
		
		public HashMap<String,CateInfo> getCateInfo(BufferedReader reader) throws IOException{
			String line=null;
			HashMap<String,CateInfo> cateInfos=new HashMap<String,CateInfo>();
			while((line=reader.readLine())!=null){
				String fields[]=line.split("\t");
				String cateid=fields[0];
				String catename=fields[2];
				CateInfo cate=new CateInfo(catename);
				String levels[]=fields[5].split(",");
				int len=levels.length<2?levels.length:2;
				for(int i=0;i<len;i++){
					cate.setCateids(i, levels[i]);
				}
				cateInfos.put(cateid, cate);
			}
			reader.close();
			return cateInfos;
		}

	}
	
	public static int hadoopMainEntrance(String statDate) throws Exception{
		statDate=DateUtil.strToStr(statDate, "yyyyMMdd");
		Configuration conf = new Configuration();		
		Path catePath = new Path("/dsap/rawdata/cmc_category/" + statDate
				+ "/part-m-00000");
		Path localPath = new Path("/dsap/rawdata/cmc_diplocal/" + statDate
				+ "/part-m-00000");
		String outPath1UriWithLink = catePath.toUri().toString() + "#"
				+ "BelongCate.part";// #号之后的名称是对上面文件的链接，不同文件的链接名必须不能相同
		String outPath2UriWithLink = localPath.toUri().toString() + "#"
				+ "DisLocal.part";
		try {
			DistributedCache.addCacheFile(new URI(outPath1UriWithLink), conf);
		} catch (URISyntaxException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		DistributedCache.addCacheFile(new URI(outPath2UriWithLink), conf);

		int returnCode = 126;
		conf.set("DATE", statDate);
		Job job = new Job(conf, "EC_Post_Detail_Job");
		job.setJarByClass(FourthInfo.class);
		job.setMapperClass(FourthInfoMapper.class);
		job.setReducerClass(FourthInfoReducer.class);
		job.setNumReduceTasks(4);
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(Text.class);
		job.setOutputKeyClass(NullWritable.class);
		job.setOutputValueClass(Text.class);
		if (FileSystem.get(conf).exists(new Path(input.replaceAll("yyyyMMdd", statDate)))) {
			FileInputFormat.addInputPath(job, new Path(input.replaceAll("yyyyMMdd", statDate)));
		} else {
			return -1;
		}
		String outPath=output.replaceAll("yyyyMMdd", statDate);
		if(checkPath(outPath)){
			deleteDir(outPath, true);
		}
		FileOutputFormat.setOutputPath(job, new Path(outPath));
		returnCode = job.waitForCompletion(true) ? 0 : 1;
		return returnCode;
	}

	 /**删HDFS文件夹
     * @param dirPath  路径
     * @param isNull	true:非空文件夹也删除   false:只删除空文件夹
     * @throws IOException
     */
    public static boolean deleteDir(String dirPath,boolean isNull) throws IOException
    {
        Configuration conf = new Configuration();
        FileSystem fs = FileSystem.get(conf);
        Path path=new Path(dirPath);
        boolean isDeleted  = fs.delete(path, isNull);
        
        fs.close();
        return isDeleted;
    }
    
    
    public static boolean checkPath(String dirPath) throws IOException
    {
        Configuration conf = new Configuration();
        FileSystem fs = FileSystem.get(conf);
        Path path=new Path(dirPath);
        boolean exists  = fs.exists(path);
        
        fs.close();
        return exists;
    }
    
	public static class LocalInfo implements Serializable{
		/**
		 * 
		 */
		private static final long serialVersionUID = 1L;
		private String localname;
		private String[] localids=new String[3];
		public LocalInfo(String localname){
			this.localname=localname;
			for(int i=0;i<3;i++)
				localids[i]="0";
		}
		public void setLocalIds(String id1,String id2,String id3){
			localids[0]=id1;
			localids[1]=id2;
			localids[2]=id3;
		}
		public String getLocalname() {
			return localname;
		}
		public void setLocalname(String localname) {
			this.localname = localname;
		}
		public String[] getLocalids() {
			return localids;
		}
		public void setLocalids(String[] localids) {
			this.localids = localids;
		}
		public void setLocalids(int index,String id) {
			localids[index] = id;
		}
	}
	public static class CateInfo implements Serializable{
		/**
		 * 
		 */
		private static final long serialVersionUID = 2L;
		private String catename;
		private String[] cateids=new String[2];
		
		public CateInfo(String catename) {
			this.catename=catename;
			for(int i=0;i<2;i++)
				cateids[i]="0";
		}
		public void setCateIds(String id1,String id2){
			cateids[0]=id1;
			cateids[1]=id2;
		}
		public String getCatename() {
			return catename;
		}
		public void setCatename(String catename) {
			this.catename = catename;
		}
		public String[] getCateids() {
			return cateids;
		}
		public void setCateids(String[] cateids) {
			this.cateids = cateids;
		}
		public void setCateids(int index,String id) {
			cateids[index] = id;
		}
	}


	/**
	 * @param args
	 * @throws Exception 
	 */
	public static void main(String[] args) throws Exception {
		String startDate = "";
		String endDate = "";
		
		if(args.length > 0 ){
			if(args.length == 1 ){
				startDate = args[0];
				endDate = args[0];
			}else if(args.length == 2 ){
				startDate = args[0];
				endDate = args[1];
			}
		}
		
		if(startDate == null || "".equals(startDate)){
			startDate = DateUtil.getStringDate(DateUtil.getDateBefore(new Date(), 1), "yyyy-MM-dd");
			endDate = startDate;
		}
		
		for(String statDate:DateUtil.getDateList(startDate, endDate)){
			System.out.println("deal:" + statDate);
            hadoopMainEntrance(statDate);
		}
	}

}
